package config_center

import (
	"context"
	"fmt"
	jsoniter "github.com/json-iterator/go"
	"github.com/zeromicro/go-queue/kq"
	"github.com/zeromicro/go-zero/core/service"
	"payme3000/framework/db"
	"payme3000/framework/util/check"

	"payme3000/framework/errorx"
	"strings"
)

type IKafkaReactor interface {
	AddProducer(topic string, producer any, brokers []string)
	AddConsumer(conf *kq.KqConf, handle any, configCenter *ConfigCenter, isCheckDuplication bool, isMutex bool)
	StartAllConsumers()
	StopAllConsumers()
	GetTopicToDbTypeMap() map[string]db.Type
}

func (c *ConfigCenter) AddProducer(topic string, producer any) error {
	dbType, ok := c.kafkaReactor.GetTopicToDbTypeMap()[topic]
	if !ok {
		return errorx.Errorf(context.Background(), "empty db type with topic:%s", topic)
	}

	// 从配置中心获取连接配置
	key := c.dbReactor.GetDbConfigKey(dbType)
	if key == "" {
		return errorx.Errorf(context.Background(), "empty db config key with topic:%s, dbType:%d", topic, dbType)
	}

	// 监听这个key
	if err := c.Watch(&ConfigItem{
		Key:   key,
		Value: NewAtomicString(""),
		NotifyFunc: func(newValue IAtomic) {
			dbConfig := &db.Config{}
			if err := jsoniter.Unmarshal([]byte(newValue.String()), dbConfig); err != nil {
				errorx.Catch(context.Background(), err)
				return
			}

			// config center主要作用是获取brokers
			c.kafkaReactor.AddProducer(topic, producer, strings.Split(dbConfig.Host, ","))
		},
	}); err != nil {
		return err
	}

	return nil
}

func (c *ConfigCenter) AddConsumer(topic string, group string, handle any, isCheckDuplication bool, isMutex bool) error {
	if topic == "" || group == "" || check.IsRealNil(handle) {
		return errorx.Errorf(context.Background(), "invalid params to add consumer with topic:%s, group:%s, handle:%d", topic, group, handle)
	}

	dbType, ok := c.kafkaReactor.GetTopicToDbTypeMap()[topic]
	if !ok {
		return errorx.Errorf(context.Background(), "empty db type with topic:%s", topic)
	}

	// 从配置中心获取连接配置
	key := c.dbReactor.GetDbConfigKey(dbType)
	if key == "" {
		return errorx.Errorf(context.Background(), "empty db config key with topic:%s, dbType:%d", topic, dbType)
	}

	// 监听这个key
	if err := c.Watch(&ConfigItem{
		Key:   key,
		Value: NewAtomicString(""),
		NotifyFunc: func(newValue IAtomic) {
			dbConfig := &db.Config{}
			if err := jsoniter.Unmarshal([]byte(newValue.String()), dbConfig); err != nil {
				errorx.Catch(context.Background(), err)
				return
			}

			conf := getDefaultConsumerConf(fmt.Sprintf("%s:%s", topic, group))
			conf.Brokers = strings.Split(dbConfig.Host, ",")
			conf.Topic = topic
			conf.Group = group

			c.kafkaReactor.AddConsumer(&conf, handle, c, isCheckDuplication, isMutex)
		},
	}); err != nil {
		return err
	}

	return nil
}

func (c *ConfigCenter) StartAllConsumers() {
	if c.kafkaReactor != nil {
		c.kafkaReactor.StartAllConsumers()
	}
}

func (c *ConfigCenter) StopAllConsumers() {
	if c.kafkaReactor != nil {
		c.kafkaReactor.StopAllConsumers()
	}
}

// getDefaultConsumerConf 消费者默认配置
func getDefaultConsumerConf(name string) kq.KqConf {
	return kq.KqConf{
		ServiceConf: service.ServiceConf{Name: name},
		Offset:      "last",
		Conns:       1,
		Consumers:   8,
		Processors:  8,
		MinBytes:    10240,
		MaxBytes:    10485760,
		ForceCommit: false,
	}
}
